package com.sunwayworld.basemodule.common.asynctask.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sunwayworld.basemodule.common.asynctask.bean.LimsAsynctaskBean;
import com.sunwayworld.basemodule.common.asynctask.bean.LimsAsynctaskInsert;
import com.sunwayworld.basemodule.common.asynctask.bean.LimsAsynctaskScheduleBean;
import com.sunwayworld.basemodule.common.asynctask.bean.LimsAsynctaskServerBean;
import com.sunwayworld.basemodule.common.asynctask.comsumer.AsyncTaskConsumer;
import com.sunwayworld.basemodule.common.asynctask.config.AsyncTaskExecutorConfig;
import com.sunwayworld.basemodule.common.asynctask.config.AsyncTaskStatusEnum;
import com.sunwayworld.basemodule.common.asynctask.config.AsyncTaskTypeEnum;
import com.sunwayworld.basemodule.common.asynctask.config.LimsAsyncTaskExecutor;
import com.sunwayworld.basemodule.common.asynctask.dao.LimsAsynctaskDao;
import com.sunwayworld.basemodule.common.asynctask.service.LimsAsynctaskScheduleService;
import com.sunwayworld.basemodule.common.asynctask.service.LimsAsynctaskServerService;
import com.sunwayworld.basemodule.common.asynctask.service.LimsAsynctaskService;
import com.sunwayworld.basemodule.common.asynctask.statics.TaskStaticsInfo;
import com.sunwayworld.basemodule.common.asynctask.statics.TaskTypeExecuteInfo;
import com.sunwayworld.basemodule.common.asynctask.util.AsyncTaskExecutorHelper;
import com.sunwayworld.basemodule.common.eptstorage.db.DynamicBeanHanderService;
import com.sunwayworld.framework.at.annotation.AuditTrailEntry;
import com.sunwayworld.framework.at.annotation.AuditTrailType;
import com.sunwayworld.framework.context.ApplicationContextHelper;
import com.sunwayworld.framework.context.concurrent.GikamConcurrentLocker;
import com.sunwayworld.framework.exception.InvalidDataException;
import com.sunwayworld.framework.mybatis.mapper.MatchPattern;
import com.sunwayworld.framework.mybatis.mapper.SearchFilter;
import com.sunwayworld.framework.restful.data.RestJsonWrapperBean;
import com.sunwayworld.framework.spring.annotation.GikamBean;
import com.sunwayworld.framework.utils.*;
import com.sunwayworld.module.mdm.user.bean.CoreUserBean;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.env.Environment;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;

import java.lang.reflect.Method;
import java.util.*;
import java.util.stream.Collectors;

@Repository
@GikamBean
public class LimsAsynctaskServiceImpl implements LimsAsynctaskService {
    private static final Logger logger = LogManager.getLogger(LimsAsynctaskService.class);
    @Autowired
    private LimsAsynctaskDao limsAsynctaskDao;
    @Autowired
    private AsyncTaskExecutorConfig asyncTaskExecutorConfig;
    @Autowired
    @Lazy
    private AsyncTaskExecutorHelper asyncTaskExecutorHelper;
    @Autowired
    private Environment env;
    @Autowired
    private AsyncTaskConsumer taskScheduleConsumer;
    @Autowired
    private LimsAsynctaskScheduleService limsAsynctaskScheduleService;
    @Autowired
    private LimsAsynctaskServerService limsAsynctaskServerService;

    @Override
    @SuppressWarnings("unchecked")
    public LimsAsynctaskDao getDao() {
        return limsAsynctaskDao;
    }

    @Override
    @Transactional
    @AuditTrailEntry(AuditTrailType.INSERT)
    public Long insert(RestJsonWrapperBean jsonWrapper) {
        LimsAsynctaskBean limsAsynctask = jsonWrapper.parseUnique(LimsAsynctaskBean.class);
        limsAsynctask.setId(ApplicationContextHelper.getNextIdentity());
        getDao().insert(limsAsynctask);
        return limsAsynctask.getId();
    }

    //加入任务只需填写关键字段taskId taskName executorName jsonParam serviceMethod taskNameDesc taskNotify taskNotifyTemplateId taskUserId taskUserName
    @Override
    @Transactional
    public Map<String, Object> addAsyncTask(List<LimsAsynctaskInsert> inserts, CoreUserBean loginUser) {
        logger.info("-----addAsyncTask---------------");
        GikamConcurrentLocker.block("addAsyncTask-Lock");
        //对于相同的任务。如果有待入队、 执行中、执行失败，则不允许加。
        List<Long> taskIds = inserts.stream().map(f -> f.getTaskId()).distinct().collect(Collectors.toList());
        List<LimsAsynctaskBean> existTasks = this.selectListByFilter(SearchFilter.instance().match("taskId", taskIds).filter(MatchPattern.OR)
                .match("execStatus", Arrays.asList(AsyncTaskStatusEnum.pendingAssign.status,AsyncTaskStatusEnum.hasAssign.status,AsyncTaskStatusEnum.executing.status,AsyncTaskStatusEnum.executeFail.status)).filter(MatchPattern.OR));

        List<String> msgs = new ArrayList<>();
        List<LimsAsynctaskBean> items = new ArrayList<>();
        for (LimsAsynctaskInsert asynctaskInsert : inserts) {
            Optional<LimsAsynctaskBean> first = existTasks.stream().filter(f -> f.getTaskId().longValue() == asynctaskInsert.getTaskId().longValue()).findFirst();
            if(first.isPresent()){
                msgs.add("任务ID:"+asynctaskInsert.getTaskId()+"的任务已提交,，请勿重复操作");
                logger.warn("任务ID:"+asynctaskInsert.getTaskId()+"的任务已提交,，请勿重复操作");
                continue;
            }

            LimsAsynctaskBean item = new LimsAsynctaskBean();
            BeanUtils.copyProperties(asynctaskInsert,item);
            item.setId(ApplicationContextHelper.getNextIdentity());
            item.setExecStatus(AsyncTaskStatusEnum.pendingAssign.status);
            items.add(item);
        }
        this.getDao().insert(items);

        if(ObjectUtils.isEmpty(msgs)){
            msgs.add("操作成功");
        }
        Map<String,Object> result = new HashMap<>();
        result.put("msg", StringUtils.join(msgs,","));
        return result;
    }

    //将宕掉的节点下除开成功的的任务，都重置
    @Transactional
    @Override
    public void initialTaskOnDeadLives() {
        List<LimsAsynctaskServerBean> serverBeans = limsAsynctaskServerService.selectAll();
        if(ObjectUtils.isEmpty(serverBeans)){
            return;
        }
        List<String> deads = new ArrayList<>();
        for (LimsAsynctaskServerBean serverBean : serverBeans) {
            String healthUrl = serverBean.getHealthUrl();
            if(ObjectUtils.isEmpty(healthUrl)){
                logger.error("healthUrl of {} is null",serverBean.getId());
                continue;
            }
            boolean live = restTemplateGetCheckLive(healthUrl);
            if(!live && !deads.contains(serverBean.getId())){
                deads.add(serverBean.getId());
            }
        }
        if(ObjectUtils.isEmpty(deads)){
            return;
        }
        logger.error("节点={}未启动",StringUtils.join(deads,","));

        logger.info("initialTaskOnDeadLives");
        List<LimsAsynctaskBean> tasksOnDeadLive = this.selectListByFilter(SearchFilter.instance().match("serverId",deads).filter(MatchPattern.OR).match("execStatus",AsyncTaskStatusEnum.executeSuccess.status).filter(MatchPattern.DIFFER));
        for (LimsAsynctaskBean asynctaskBean : tasksOnDeadLive) {
            asynctaskBean.setServerId(null);
            asynctaskBean.setExecStatus(AsyncTaskStatusEnum.pendingAssign.status);
            asynctaskBean.setQueueProgress(null);
            asynctaskBean.setErrMsg(null);
            asynctaskBean.setExecStartDate(null);
            asynctaskBean.setExecEndDate(null);
            asynctaskBean.setJoinedDate(null);
            asynctaskBean.setExecCount(0L);
        }
        this.getDao().update(tasksOnDeadLive,"ServerId","ExecStatus","QueueProgress","ErrMsg","ExecStartDate","ExecEndDate","JoinedDate","ExecCount");
        logger.warn("宕机节点={},未成功的任务数={},重置完成",StringUtils.join(deads,","),tasksOnDeadLive.size());
    }

    @Override
    public boolean restTemplateGetCheckLive(String interfaceUrl) {
        RestTemplate restTemplate;
        if(interfaceUrl.contains("https")){
            restTemplate = (RestTemplate) ApplicationContextHelper.getBean(DynamicBeanHanderService.class).getBean("httpsRestTemplate");
        }else{
            restTemplate = (RestTemplate) ApplicationContextHelper.getBean(DynamicBeanHanderService.class).getBean("httpRestTemplate");
        }
        try {
            ResponseEntity<String> responseEntity = restTemplate.getForEntity(interfaceUrl, String.class);
            if(responseEntity!=null && responseEntity.getStatusCode()== HttpStatus.OK){
                return true;
            }else{
                logger.error("接口={},失败",interfaceUrl);
            }
        }catch (Exception s){
            logger.error("接口={},失败",interfaceUrl);
        }
        return false;
    }

    @Override
    public void scheduleUpdateWaitExecTask() {
/*        if(!asyncTaskExecutorConfig.getEnable()){
            return;
        }
        logger.info("-----updateWaitAsyncTaskWaitNum start---------------");
        List<AsyncTaskExecutorConfig.ClusterServer> clusterServers = asyncTaskExecutorConfig.getClusterServers();
        List<String> ids = clusterServers.stream().map(f -> f.getId()).distinct().collect(Collectors.toList());
        asyncTaskExecutorHelper.updateWaitAsyncTaskWaitNum(ids);
        logger.info("-----updateWaitAsyncTaskWaitNum end---------------");*/
    }

    @Override
    public Map<String,Object> testTask(List<LimsAsynctaskBean> limsAsynctaskBeans) {
        //测试配置的task是否正确，强制在当前节点执行
        Map<String,Object> result = new HashMap<>();
        String msg = "success";
        for (LimsAsynctaskBean limsAsynctaskBean : limsAsynctaskBeans) {
            if(limsAsynctaskBean.getTaskId()==null){
                throw new InvalidDataException("任务ID不能为空！");
            }
            if(ObjectUtils.isEmpty(limsAsynctaskBean.getTaskName())){
                throw new InvalidDataException("任务名称不能为空！");
            }
            if(ObjectUtils.isEmpty(limsAsynctaskBean.getTaskType())){
                throw new InvalidDataException("任务类型不能为空！");
            }
            if(ObjectUtils.isEmpty(limsAsynctaskBean.getExecutorName())){
                throw new InvalidDataException("任务执行器名称不能为空！");
            }
            List<LimsAsynctaskScheduleBean> limsAsynctaskScheduleBeans = limsAsynctaskScheduleService.selectAll();
            if(ObjectUtils.isEmpty(limsAsynctaskScheduleBeans)){
                throw new InvalidDataException("未配置异步任务线程池！");
            }
            boolean present = limsAsynctaskScheduleBeans.stream().filter(c -> limsAsynctaskBean.getExecutorName().equals(c.getExecutorId())).findFirst().isPresent();
            if(!present){
                List<String> list = limsAsynctaskScheduleBeans.stream().map(f -> f.getExecutorId()).distinct().collect(Collectors.toList());
                throw new InvalidDataException("任务执行器不存在，只能输入下面这些执行器:"+ StringUtils.join(list,","));
            }
            if(!ObjectUtils.isEmpty(limsAsynctaskBean.getJsonParam())){
                try {
                    logger.info("JsonParam:{}",limsAsynctaskBean.getJsonParam());
                    JSON.parseObject(limsAsynctaskBean.getJsonParam());
                }catch (Throwable s){
                    s.printStackTrace();
                    throw new InvalidDataException("请输入JSON格式的参数！");
                }
            }
            if(ObjectUtils.isEmpty(limsAsynctaskBean.getServiceMethod())){
                throw new InvalidDataException("调用方法不能为空！");
            }
            int index = limsAsynctaskBean.getServiceMethod().lastIndexOf(".");
            if(index==-1){
                throw new InvalidDataException("请输入调用方法的全路径！");
            }
            String serviceName = limsAsynctaskBean.getServiceMethod().substring(0, index);
            String methodName = limsAsynctaskBean.getServiceMethod().substring(index + 1);
            Class<?> serviceClazz = null;
            try {
                serviceClazz = ClassUtils.getClass(serviceName);
            }catch (Throwable s){
                s.printStackTrace();
            }
            if(serviceClazz==null){
                throw new InvalidDataException("调用class不存在！");
            }
            Object service = ApplicationContextHelper.getBean(serviceClazz);
            if(service==null){
                throw new InvalidDataException("调用class不存在！");
            }
            Method method;
            if(!ObjectUtils.isEmpty(limsAsynctaskBean.getJsonParam())){
                //methodName 如果有参数，参数必须RestJsonWrapperBean
                method = ReflectionUtils.findMethod(serviceClazz, methodName, new Class[]{JSONObject.class});
            }else{
                method = ReflectionUtils.findMethod(serviceClazz, methodName);
            }
            if(method==null){
                throw new InvalidDataException("调用方法不存在！");
            }
        }
        taskScheduleConsumer.executeTaskOnCurrentNode(limsAsynctaskBeans);
        result.put("msg",msg);
        return result;
    }

    @Override
    public void testWithJsonParam(JSONObject jsonObject) {
        logger.info("execute testJsonParam={}",jsonObject);
        //直接抛异常，阻断式，可以记录异常
        if(true){
            throw new InvalidDataException("报告错误码:2000");
        }
    }
    @Override
    public String testWithOutJsonParam() {
        List<String> warns = new ArrayList<>();
        //非阻断式，记录一些报告生成中的告警信息
        int sleep = new Random().nextInt((20 - 5) + 1) + 5;
        try {
            Thread.sleep(sleep*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        warns.add("soc无数据");
        warns.add("样品条码1错误");
        return StringUtils.join(warns,",");
    }

    @Override
    public TaskStaticsInfo queryTaskStaticsInfo() {
        return this.getDao().queryTaskStaticsInfo();
    }

    @Override
    public TaskTypeExecuteInfo queryTaskTypeExecuteInfo() {
        TaskTypeExecuteInfo taskTypeExecuteInfo = new TaskTypeExecuteInfo();
        List<String> statusDescList = new ArrayList<>();
        AsyncTaskStatusEnum[] enums = AsyncTaskStatusEnum.values();
        for (AsyncTaskStatusEnum anEnum : enums) {
            statusDescList.add(anEnum.statusDesc);
        }
        taskTypeExecuteInfo.setStatusList(statusDescList);

        List<TaskTypeExecuteInfo.Source> sources = new ArrayList<>();
        AsyncTaskTypeEnum[] asyncTaskTypeEnums = AsyncTaskTypeEnum.values();
        List<Map<String, Object>> mapList = getDao().queryTaskTypeStatusStatics();
        for (AsyncTaskTypeEnum asyncTaskTypeEnum : asyncTaskTypeEnums) {
            TaskTypeExecuteInfo.Source e = new TaskTypeExecuteInfo.Source();
            e.setTaskType(asyncTaskTypeEnum.desc);
            List<Integer> countList = new ArrayList<>();
            for (AsyncTaskStatusEnum statusEnum : enums) {
                int num = 0;
                //求取不同任务类型下不同状态的数量
                Optional<Map<String, Object>> first = mapList.stream().filter(f -> asyncTaskTypeEnum.desc.equals(f.get("TASKTYPE")) && statusEnum.status.equals(f.get("EXECSTATUS"))).findFirst();
                if(first.isPresent()){
                    num = Integer.parseInt(ObjectUtils.toString(first.get().get("NUM")));
                }
                countList.add(num);
            }
            e.setCountlist(countList);
            sources.add(e);
        }

        taskTypeExecuteInfo.setSources(sources);
        return taskTypeExecuteInfo;
    }

    @Override
    public List<LimsAsynctaskServerBean> asyncServerHealth() {
        List<LimsAsynctaskServerBean> asynctaskServerBeans = limsAsynctaskServerService.selectAll();
        for (LimsAsynctaskServerBean asynctaskServerBean : asynctaskServerBeans) {
            if(!ObjectUtils.isEmpty(asynctaskServerBean.getHealthUrl())){
                boolean b = restTemplateGetCheckLive(asynctaskServerBean.getHealthUrl());
                asynctaskServerBean.setExt$Item("health",b?"正常":"异常");
            }
        }
        return asynctaskServerBeans;
    }

    @Override
    public List<LimsAsynctaskScheduleBean> asyncServerScheduleStatus(RestJsonWrapperBean wrapperBean) {
        String serverId = wrapperBean.getParamValue("serverId");
        List<LimsAsynctaskScheduleBean> asynctaskScheduleBeans = limsAsynctaskScheduleService.selectListByFilter(SearchFilter.instance().match("serverId", serverId).filter(MatchPattern.SEQ)
                .match("enable", "1").filter(MatchPattern.SEQ));
        for (LimsAsynctaskScheduleBean asynctaskScheduleBean : asynctaskScheduleBeans) {
            String executorId = asynctaskScheduleBean.getExecutorId();
            LimsAsyncTaskExecutor executor = asyncTaskExecutorHelper.getByExecutorName(executorId);
            if(executor!=null){
                int waitQueueSize = executor.getQueue().size();
                int activeCount = executor.getActiveCount();
                int maximumPoolSize = executor.getMaximumPoolSize();
                logger.info("{}:maximumPoolSize={},当前排队任务数:{},当前进行中任务数:{}",executorId,maximumPoolSize,waitQueueSize,activeCount);
                asynctaskScheduleBean.setExt$Item("activecount",activeCount+"");
                asynctaskScheduleBean.setExt$Item("waitqueuesize",waitQueueSize+"");
            }else{
                asynctaskScheduleBean.setExt$Item("activecount","-1");
                asynctaskScheduleBean.setExt$Item("waitqueuesize","-1");
            }
        }
        return asynctaskScheduleBeans;
    }

}
